package cn.mq;

import com.ibm.mq.MQException;
import com.ibm.mq.MQQueue;
import com.ibm.mq.MQQueueManager;
import com.ibm.msg.client.wmq.v6.base.internal.MQC;


import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;

import static cn.mq.InsertIntoMysqlUtil.*;

import org.apache.log4j.Logger;

public class MQReceiver {

    private static final Logger logger = Logger.getLogger(MQReceiver.class); // 创建 Logger 实例

    private static final int THREAD_COUNT = 10;  // 线程数

    private static final BlockingQueue<String> messageQueue = new LinkedBlockingQueue<>();  // 用于存放消息的阻塞队列

    public static void main(String[] args) {

        // 启动监控线程
        new Thread(() -> monitorQueue("QUEUE")).start();


        // 启动线程池，处理消息
        ExecutorService executorService = Executors.newFixedThreadPool(THREAD_COUNT);
        for (int i = 0; i < THREAD_COUNT; i++) {
            executorService.submit(new MessageProcessor());  // 处理消息的工作线程
        }
    }
    // 监控队列的消息数量
    public static void monitorQueue(String queueName) {
        MQQueueManager mqQueueManager = null;
        MQQueue queue = null;
        try {
            // 连接到队列管理器
            mqQueueManager = MQUtil.getMQQueueMgr(MQUtil.getPropertyContext());
            int openOptions = MQC.MQOO_INQUIRE | MQC.MQOO_INPUT_SHARED;

            // 打开队列
            queue = mqQueueManager.accessQueue(queueName, openOptions, null, null, null);

            while (true) {  // 持续监控
                int currentDepth = queue.getCurrentDepth();  // 获取当前队列深度
                logger.info("当前队列深度" + currentDepth + "  ");

                if (currentDepth > 0) {  // 如果队列中消息增加
                    logger.info("检测到新消息，开始读取...");
                    String msgStrJson = MQUtil.receiverMessageToMQ(queueName);  // 读取消息
                    messageQueue.offer(msgStrJson);  // 将消息放入阻塞队列
                }

                Thread.sleep(5000 * 30);  // 每隔5秒检查一次队列
            }

        } catch (MQException | InterruptedException e) {
            logger.error("监控队列时出错", e); // 记录错误信息
            // 发送邮件
            String errorMessage = "监控队列时出错: " + e.getMessage();
            EmailUtil.sendMail("liuzhili@unicdata.com" , "MQ监控异常" , errorMessage);
        } finally {
            MQUtil.closeMQQueueManager(queue, mqQueueManager);  // 确保资源关闭
        }
    }

    // 消息处理器，处理业务逻辑的线程
    static class MessageProcessor implements Runnable {
        @Override
        public void run() {
            try {
                while (true) {
                    String msg = messageQueue.take();  // 从阻塞队列中获取消息
                    try {
                        insertSendElectronicInvoice(msg);  // 处理消息，插入数据库
                    } catch (Exception e) {
                        logger.error("消息处理时出错" , e); // 记录错误信息
                        // 发送邮件
                        String errorMessage = "处理消息时出错: " + e.getMessage() + "\n消息内容: " + msg;

                        EmailUtil.sendMail("liuzhili@unicdata.com" ,"消息处理异常" , errorMessage);
                    }
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();  // 恢复线程中断状态
                e.printStackTrace();
            }
        }
    }
}
